/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package unisb.cs.core.join.cogroup.index;

import java.io.DataOutputStream;
import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Progressable;

public class CogroupIndexerOutputFormat extends TextOutputFormat<Text, Text> {

	/**
	 * Write only the values (not the keys) emitted from the reducer.
	 *
	 */
	static class RelationRecordWriter extends LineRecordWriter<Text, Text> {
		public RelationRecordWriter(DataOutputStream out, JobConf conf) {
			super(out);
		}

		/**
		 * Write output value
		 */
		public synchronized void write(Text key, Text value) throws IOException {
			if (value != null) {
				out.write(value.getBytes(), 0, value.getLength());
			}
		}

		public void close() throws IOException {
			super.close(null);
		}
	}

	public RecordWriter<Text, Text> getRecordWriter(FileSystem ignored, JobConf job, String name, Progressable progress) throws IOException {
		Path dir = getWorkOutputPath(job);
		FileSystem fs = dir.getFileSystem(job);
		FSDataOutputStream fileOut = fs.create(new Path(dir, name), progress);
		return new RelationRecordWriter(fileOut, job);
	}
}